package blue.mqtt4j.server.net;

import blue.mqtt4j.core.handler.HeartbeatHandler;
import blue.mqtt4j.server.config.Config;
import blue.mqtt4j.server.config.ConfigItem;
import blue.mqtt4j.server.topic.TopicChannel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.mqtt.MqttDecoder;
import io.netty.handler.codec.mqtt.MqttEncoder;
import io.netty.handler.timeout.IdleStateHandler;

/**
 * @author Jin Zheng
 * @since 2019-05-02
 */
public class ServerInitializer extends ChannelInitializer<SocketChannel>
{
	private final int timeout;
	private final int heartbeat;
	private final TopicChannel topicChannel;

	public ServerInitializer()
	{
		Config config = Config.getInstance();
		this.timeout = config.getInt(ConfigItem.SESSION_TIMEOUT, ConfigItem.SESSION_TIMEOUT_VALUE);
		this.heartbeat = config.getInt(ConfigItem.SESSION_HEARTBEAT, ConfigItem.SESSION_HEARTBEAT_VALUE);
		this.topicChannel = new TopicChannel();
	}

	@Override
	protected void initChannel(SocketChannel ch) throws Exception
	{
		ChannelPipeline cp = ch.pipeline();
		cp.addLast(new MqttDecoder());
		cp.addLast(MqttEncoder.INSTANCE);
		cp.addLast(new IdleStateHandler(timeout, heartbeat, heartbeat));
		cp.addLast(new HeartbeatHandler());
		cp.addLast(new MqttConnectHandler());
		cp.addLast(new MqttPublishHandler(topicChannel));
		cp.addLast(new MqttSubscribeHandler(topicChannel));
	}

}
